package org.devgframwork.datasync.core.support.executors;


import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestResult;
import io.searchbox.core.DeleteByQuery;
import lombok.extern.slf4j.Slf4j;
import org.devgframwork.datasync.core.entity.TargetConfig;
import org.devgframwork.datasync.core.entity.TriggerConfig;
import org.devgframwork.datasync.core.model.ExecutorModel;
import org.devgframwork.datasync.core.model.ResultModel;
import org.devgframwork.datasync.core.support.DataSourceMemory;
import org.devgframwork.datasync.core.support.parser.HandleParser;
import org.devgframwork.datasync.core.support.parser.impl.SqlDefaultParser;
import org.devgframwork.datasync.core.utils.CommonUtil;
import org.devgframwork.datasync.es56.executors.Es56Executor;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.List;
import java.util.concurrent.*;

/**
 * {"CONNECTION_NAME":"local_dev","TABLE_NAME":"tb_merchandise","store_prod_no":"1557613320071639043"}
 */
@Slf4j
public class TriggerExecutor implements AbstractExecutor<String> {


    public static final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            500,
            5000,
            30L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(2000),
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "data-sync, admin JdbcToJdbcExecutor-threadPoolExecutorThreadPool-" + r.hashCode());
                }
            },
            new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    r.run();
                    log.warn(">>>>>>>>>>> data-sync, registry or remove too fast, match threadpool rejected handler(run now).");
                }
            });

    @Override
    public ResultModel invoke(ExecutorModel<String> executorModel) {
        String data = executorModel.getExt();
        if (CommonUtil.isNullOrEmpty(data)) {
            return ResultModel.newFail("data cannot by empty");
        }
        JSONObject jsonObject = JSONObject.parseObject(data);
        String connectionName = jsonObject.getString("CONNECTION_NAME");
        String tableName = jsonObject.getString("TABLE_NAME");
        if (CommonUtil.isNullOrEmpty(connectionName) || CommonUtil.isNullOrEmpty(tableName)) {
            return ResultModel.newFail("connectionName and tableName cannot by empty");
        }
        String message = "connectionName:".concat(connectionName).concat(" and tableName:").concat(tableName);
        String mapKey = connectionName.concat(":").concat(tableName);
        TriggerConfig triggerConfig = DataSourceMemory.triggerConfigMap.get(mapKey);
        if (null == triggerConfig) {
            String erroMessage = message.concat("triggerConfigMap cannot by empty");
            return ResultModel.newFail(erroMessage);
        }
        String esDelete = (jsonObject.get("es_delete") == null ? "0" : jsonObject.get("es_delete").toString());
        if (CommonUtil.eq(esDelete, "1")) {
            invokeEsDelete(triggerConfig, jsonObject);
        }
        Integer isDelete = triggerConfig.getIsDelete();
        if (null != isDelete && isDelete.equals(1)) {
            String erroMessage = message.concat("is delete");
            return ResultModel.newFail(erroMessage);
        }
        if (CommonUtil.isNullOrEmpty(triggerConfig.getTargetInfo())) {
            String erroMessage = message.concat("target info cannot be empty");
            return ResultModel.newFail(erroMessage);
        }
        List<TargetConfig> targetConfigList = JSONObject.parseArray(triggerConfig.getTargetInfo(), TargetConfig.class);
        JdbcTemplate sourceJdbcTemplate = DataSourceMemory.jdbcTemplateMap.get(triggerConfig.getSourceKey());
        String sourceContentSql = DataSourceMemory.sqlMap.get(triggerConfig.getSourceContent());
        HandleParser handleParser = new SqlDefaultParser(sourceContentSql, jsonObject);
        String sqlText = handleParser.getText();
        int page = 0;
        int batchSize = Integer.MAX_VALUE;
        CompletableFuture<ResultModel> future = CompletableFuture.supplyAsync(() -> {
            ExecutorCore executorCore = new ExecutorCore(page, batchSize, sqlText, sourceJdbcTemplate, null, triggerConfig, targetConfigList, jsonObject);
            return executorCore.call();
        }, threadPoolExecutor);
        ResultModel resultModel;
        try {
            resultModel = future.get();
        } catch (Exception ex) {
            resultModel = ResultModel.newFail(ex.getMessage());
        }
        return resultModel;
    }


    private ResultModel invokeEsDelete(TriggerConfig triggerConfig, JSONObject jsonObject) {
        String esDeleteInfo = triggerConfig.getEsDeleteInfo();
        if (CommonUtil.isNullOrEmpty(esDeleteInfo)) {
            return ResultModel.newFail("ES_DELETE_INFO cannot be empty...");
        }
        String esDeleteBody = triggerConfig.getEsDeleteBody();
        HandleParser handleParser = new SqlDefaultParser(esDeleteBody, jsonObject);
        String esDeleteBodyString = handleParser.getText();
        List<TargetConfig> targetConfigList = JSONObject.parseArray(esDeleteInfo, TargetConfig.class);
        if (null == targetConfigList || targetConfigList.size() == 0) {
            return ResultModel.newFail("ES_DELETE_INFO cannot be empty...");
        }
        for (TargetConfig targetConfig : targetConfigList) {
            String targetKey = targetConfig.getTargetKey();
            if (CommonUtil.isNullOrEmpty(targetKey)) {
                continue;
            }
            String targetContent = targetConfig.getTargetContent();
            if (CommonUtil.isNullOrEmpty(targetContent)) {
                continue;
            }
            String[] indexInfo = targetContent.split("\\/");
            try {
                JestClient jestClient = Es56Executor.es56Map.get(targetKey);
                DeleteByQuery deleteByQuery = new DeleteByQuery.Builder(esDeleteBodyString).addIndex(indexInfo[0]).addType(indexInfo[1]).build();
                JestResult jestResult = jestClient.execute(deleteByQuery);
                log.info("ES_DELETE_INFO deleteByQuery:{},isSucceeded:{},responseCode:{},errorMessage:{},jsonString:{}", deleteByQuery.toString(), jestResult.isSucceeded(), jestResult.getResponseCode(), jestResult.getErrorMessage(), jestResult.getJsonString());
            } catch (Exception ex) {
                log.error(ex.getMessage(), ex);
            }
        }
        return ResultModel.newSuccess();
    }
}
